package cn.py.boot_lua.asyn;

import cn.py.boot_lua.util.exception.BusinessException;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.function.Consumer;

/***
  * @Message 异步任务工厂
 */
@Slf4j
public class AsyncTaskFactory {


    // 任务队列容器：每个任务名绑定一个引擎
    private static final Map<String, AsyncQueueTaskEngine<?>> engineMap = new ConcurrentHashMap<>();


    /**
     * 提交任务（支持自动创建并注册异步队列）
     *
     * @param taskName       任务名（唯一标识）
     * @param task           任务数据
     * @param executor       执行线程池
     * @param maxConcurrency 最大并发数，<=0表示不限制
     * @param taskHandler    执行逻辑
     * @param queueCapacity  队列容量，默认建议100
     * @param <T>            任务类型
     */
    public static <T> void submit(String taskName,
                                  T task,
                                  Executor executor,
                                  int maxConcurrency,
                                  Consumer<T> taskHandler,
                                  int queueCapacity) {
        if (taskName == null || taskName.isEmpty()) {
            throw new BusinessException("任务名不能为空");
        }
        if (taskHandler == null) {
            throw new BusinessException("任务处理逻辑不能为空");
        }
        if (executor == null) {
            throw new BusinessException("线程池不能为空");
        }

        // 注册或获取已存在队列
        AsyncQueueTaskEngine<T> engine = getOrCreateEngine(taskName, queueCapacity, executor, maxConcurrency, taskHandler);

        try {
            engine.submitTask(task);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("提交任务失败：" +  e);
        }
    }


    // 获取或注册新的任务引擎
    public static synchronized <T> AsyncQueueTaskEngine<T> getOrCreateEngine(
            String name,
            int queueCapacity,
            Executor executor,
            int maxConcurrency,
            Consumer<T> taskHandler) {

        if (engineMap.containsKey(name)) {
            return (AsyncQueueTaskEngine<T>) engineMap.get(name);
        }
        AsyncQueueTaskEngine<T> engine = new AsyncQueueTaskEngine<>(queueCapacity, executor, maxConcurrency, taskHandler);
        engineMap.put(name, engine);
        return engine;
    }

    /**
     * 关闭某个任务队列
     */
    public static void shutdown(String taskName) {
        AsyncQueueTaskEngine<?> engine = engineMap.remove(taskName);
        if (engine != null) {
            engine.shutdown();
            log.info("任务队列已关闭: {}", taskName);
        }
    }

    /**
     * 关闭所有任务队列（应用关闭时调用）
     */
    public static void shutdownAll() {
        for (Map.Entry<String, AsyncQueueTaskEngine<?>> entry : engineMap.entrySet()) {
            try {
                entry.getValue().shutdown();
                log.info("关闭队列: {}", entry.getKey());
            } catch (Exception e) {
                log.error("关闭队列异常: {}", entry.getKey(), e);
            }
        }
        engineMap.clear();
    }
}